package com.netty.client;

import com.alibaba.fastjson.JSONObject;
import com.topsoft.common.MiscHelper;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLDecoder;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

//WebSocketClientHandler
public abstract class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
    public Logger logger = LogManager.getLogger("【websocket.client】"+getClass().getName());

    public QueueType queueType;
    public static Boolean isPrintMsg=false;
    private Channel channel;
    public String sockId="";
    private int overLengthIsCompress=0;//消息超过多长就压缩
    private WebSocketClient webSocketClient;
    private WebSocketClientHandshaker handshaker;
    private ChannelPromise handshakeFuture;
    private Throwable cause;


    //最大可用的CPU核数
    public static final int PROCESSORS=Runtime.getRuntime().availableProcessors();
    //线程最大的空闲存活时间，单位为秒
    public static final int KEEPALIVETIME=60;
    //任务缓存队列长度
    public static final int BLOCKINGQUEUE_LENGTH=5000;
    public static ThreadPoolExecutor executor;
    public ThreadPoolExecutor getExecutor(){
        if (executor == null ) {
//            executor = new ThreadPoolExecutor(
//                    10,
//                    20,
//                    60, TimeUnit.MILLISECONDS,
//                    new ArrayBlockingQueue<Runnable>(10000),
//                    Executors.defaultThreadFactory(),
//                    new ThreadPoolExecutor.AbortPolicy());

//            executor= new ThreadPoolExecutor(
//                    PROCESSORS * 2,
//                    PROCESSORS * 4,
//                    KEEPALIVETIME,TimeUnit.SECONDS,
//                    new ArrayBlockingQueue<Runnable>(BLOCKINGQUEUE_LENGTH)
//            );
            executor = new ThreadPoolExecutor(
                    PROCESSORS * 2
                    , PROCESSORS * 4
                    , KEEPALIVETIME //线程最大空闲时间
                    , TimeUnit.SECONDS
                    , new ArrayBlockingQueue<>(BLOCKINGQUEUE_LENGTH)//线程等待队列
                    , ((r, executor2) -> {
                        logger.warn("被拒绝任务为 {}", r.toString());
                        logger.warn("WebSocket线程池 {}", executor.toString());
                        logger.warn("WebSocket线程池 {}", executor.toString());
                    })
            );
            executor.prestartAllCoreThreads(); // 预启动所有核心线程

            logger.info("WebSocketClientHandler 线程池创建完毕");
        }

        return executor;
    }

    public final void init(WebSocketClient webSocketClient, WebSocketClientHandshaker handshaker) {
        this.webSocketClient = webSocketClient;
        this.handshaker=handshaker;
        this.queueType=webSocketClient.queueType;
    }

    public final ChannelFuture handshakeFuture() {
        return handshakeFuture;
    }

    @Override
    public final void handlerAdded(ChannelHandlerContext ctx) {
        handshakeFuture = ctx.newPromise();
        channel=ctx.channel();
    }

    @Override
    /**
     * 当Channel变成活跃状态时被调用；Channel是连接/绑定、就绪的
     */
    public final void channelActive(ChannelHandlerContext ctx) {
        channel=ctx.channel();
        handshaker.handshake(channel);
        webSocketClient.channel=channel;
//        WebSocketClient.channels.put(getQueue(),channel);
        onOpen();
    }

    /**
     * 首次连接不通服务器；连接通后 断开连接，会调用此方法
     * @param ctx
     * @throws Exception
     */
    public final void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
        onClose(cause);
        //首次连接不通，会执行此方法
    }

    @Override
    /**
     * 当Channel离开活跃状态，不再连接到某个远端时被调用
     */
    public final void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //服务器停止时执行
        EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.schedule(new Runnable() {
            @Override
            public void run() {
                webSocketClient.createBootstrap(new Bootstrap(), eventLoop);
            }
        }, webSocketClient.getReConnectInterval(), TimeUnit.SECONDS);
        super.channelInactive(ctx);
    }

    public final void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.fireExceptionCaught(cause);
        this.cause=cause;
        //onfail(ctx.channel(),cause);
    }

    protected final void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception{
        channel=ctx.channel();
        if (!handshaker.isHandshakeComplete()) {
            handshaker.finishHandshake(channel, (FullHttpResponse) msg);
            //logger.info("连接成功 "+webSocketClient);
            handshakeFuture.setSuccess();
            return;
        }

        if (msg instanceof FullHttpResponse) {
            FullHttpResponse response = (FullHttpResponse) msg;
            throw new IllegalStateException("Unexpected FullHttpResponse (status="
                    + response.status() + ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
        }

        ByteBuf byteBuf; byte[] bytes;
        String msgString;
        if (msg instanceof TextWebSocketFrame){
            msgString=((TextWebSocketFrame)msg).text();
            //logger.info("收到消息 : ",msgString);
            if (msgString.startsWith("__gzip:")) {
                msgString=msgString.substring(7);
                msgString= MiscHelper.gunzip(msgString,"UTF-8");//解压
                if (msgString.startsWith("__encode:")) {
                    msgString=msgString.substring(9);
                    try {
                        msgString = URLDecoder.decode(msgString, "utf-8");
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
            if (MiscHelper.isNull(msgString)==false) {
                beforeMessage(msgString);//.content().toString(CharsetUtil.UTF_8));
            }
        }else if (msg instanceof BinaryWebSocketFrame) {//二进制消息
            byteBuf = ((BinaryWebSocketFrame)msg).content();
            bytes = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(bytes);                      //二进制数据
            msgString= MiscHelper.gunzip(bytes,"UTF-8");//解压得到的字符串
            if (MiscHelper.isNull(msgString)==false) {
                beforeMessage(msgString);
            }
        }else{
            logger.info("WebSocket Client received closing");
            channel.close();
        }
    }

//    /**
//     *  消费者 消费一条消息后(消息 __code 须有值)，如果onMessage返回true, 则会向生产者发送一条消费后确认消息，
//     *  类库会自动生成 确认消息 的返回内容，这个方法可以再处理消息的内容
//     * @param msgConfirm 确认消息
//     * @param msgConfirm 接收的消息
//     */
//    public abstract void  initConfirmMessage(JSONObject msgConfirm,JSONObject msgSource);
//
//    private void  sendMessageConfirm2(JSONObject msgConfirm,JSONObject msgSource){
//        initConfirmMessage(msgConfirm,msgSource);
//        sendMessage(msgConfirm);
//    }

    /**
     * 本队列广播消息
     * @param msgJson 消息
     */
    public void  sendMessage(JSONObject msgJson){
        sendMessage(null,msgJson);
    }
    /**
     * @param msgJson 消息
     * @param __queue  指定的接收消息的队列名  synchronized
     */
    public void sendMessage(String __queue, JSONObject msgJson){
        msgJson.put("__queue",__queue!=null&& __queue.length()>0?__queue:getQueue());
        //channel.writeAndFlush(new TextWebSocketFrame(msgJson.toJSONString()));// 发送文本消息
        String msgString=msgJson.toJSONString();
        if(msgString.length()>overLengthIsCompress){
            msgString="__gzip:"+MiscHelper.gzip2String(msgString,"UTF-8");
            channel.writeAndFlush(new TextWebSocketFrame(msgString));
            //channel.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(MiscHelper.gzip2byte(msgString,"UTF-8"))));
        }else {
            channel.writeAndFlush(new TextWebSocketFrame(msgString));
        }
    }

    private final void  beforeMessage(String msgString) {
        //多线程接处理消息
        Runnable runnable =new RunnableHandler(this,msgString);
        getExecutor().execute(runnable);

//        JSONObject msgJson;
//        try {
//            msgJson = JSONObject.parseObject(msgString);
//            String __type=msgJson.getString("__type");
//            String __code=msgJson.getString("__code");
//            String number=msgJson.getString("number");
//            JSONObject __heads=msgJson.getJSONObject("__heads");
//
//            if (__type!=null && __type.equals("success")) {
//                this.sockId = msgJson.getString("sockId");
//                onSuccess(msgJson);
//            }else if (__type!=null && (__type.equals("online") || __type.equals("offline"))) {
//                this.sockId = msgJson.getString("sockId");
//                onNotice(msgJson);
//            }else if (__type!=null && __type.equals("confirm")){
//                //生产者接收 消费者 返回的消费确认信息
//                onMessageByConfirm(msgJson,msgString);
//            }else if (__type!=null && __type.equals("error")){
//                //返回服务器发送的错误消息
//                onMessageByError(msgJson,msgString,number);
//            }else {
//                boolean ret2=__code!=null && __code.length()>0;// && (queueType==QueueType.C_ONE || queueType==QueueType.C_MORE);
//                JSONObject msgConfirm=null;
//                if (ret2){
//                    msgConfirm=new JSONObject(true);
//                    //消费者 返回消费确认信息
//                    msgConfirm.put("__type","confirm");
//                    msgConfirm.put("number","S100");
//                    msgConfirm.put("__code",__code);
//                    if(__heads!=null) {
////                        msgConfirm.put("pSockId", __heads.getString("pSockId"));//producerId=消息的发送者Id
////                        msgConfirm.put("pUserId", __heads.getString("pSockId"));//producerId=消息的发送者Id
//                        msgConfirm.put("__heads", __heads);//消息生产者的头信息
//                    }
//                }
//                boolean ret=onMessage(msgJson, msgString,msgConfirm);//先执行方法消费
//                if (ret && ret2 && msgConfirm!=null){
//                    sendMessage(msgConfirm);//发送确认消息
//                }
//                if(msgConfirm!=null) msgConfirm.clear();msgConfirm=null;
//            }
//            if(msgJson!=null)msgJson.clear();msgJson=null;msgString=null;
//        } catch (Exception e) {
//            e.printStackTrace();
//            logger.error("WebSocketClientHandler.beforeMessage 接收到的消息"+msgString);
//            logger.error("WebSocketClientHandler.beforeMessage 错误"+e.getMessage());
//            //onMessage(null,msgString);
//        }

    }

    /**
     * 接收普通消息
     * @param msgJson    接收到 JSONObject类型 的消息
     * @param msgString  接收到 String格式 的消息
     * @param msgConfirm 消费确认消息
     *
     * return boolean  消费者使用，true将发送消费确认消息给生产者   false 不发送。仅仅是消息生产者，这可以始终返回false
     */
    public abstract boolean onMessage(JSONObject msgJson, String msgString,JSONObject msgConfirm);

    /**
     * 生产者 接收 消费者消费后的确认消息。
     * @param msgJson   JSONObject类型的消息
     * @param msgString String格式的消息
     */
    public abstract void onMessageByConfirm(JSONObject msgJson, String msgString);

    /**
     * number 为错误的编号。__type=error的消息都调用这个方法
     * 接收服务器发送的错误消息，如
     * 1. 连接参数错误，连接地址没有指定 queue或queeType参数指定错误；
     * 2. 连接参数错误，连接地址queeType参数指定值错误；
     * 3. 通过__code对消息编码后，发送一次消费确认后，再次发送出现的错误
     *
     * @param msgJson   JSONObject类型的消息
     * @param msgString String格式的消息
     * @param number    String 错误编号
     */
    public abstract void onMessageByError(JSONObject msgJson, String msgString,String number);

    /**
     * 当前连接成功服务器返回的消息
     * @param msgJson JSONObject类型的消息
     */
    public void onSuccess(JSONObject msgJson){
        logger.debug(msgJson.toJSONString());
    }
    /**
     * 其他用户上线  下线的通知消息
     * @param msgJson JSONObject类型的消息
     */
    public abstract void onNotice(JSONObject msgJson);


    public void onOpen(){}
    public abstract void onClose(Throwable cause);
    /**
     * 获得连接参数
     * @return
     */
    public JSONObject getParameters(){
        return this.webSocketClient.getParameters();
    }
    /**
     * 获得队列名称
     * @return
     */
    public String getQueue(){
        return this.webSocketClient.getQueue();
    }

    public String getUrl(){
        return this.webSocketClient.getUrl();
    }
    public URI getURL(){
        return this.webSocketClient.getURL();
    }

    public final String getChannelId(){
        return this.sockId;
    }

//    private boolean isConfirm=true;
//    public boolean getIsConfirm(){
//        return this.isConfirm;
//    }
//    public void setIsConfirm(boolean isConfirm){
//        this.isConfirm=isConfirm;
//    }
}